#include "thread_pool.h"
#include "mem_alloc.h"
#include "dlist.h"
#include "log.h"

#include <assert.h>
#include <pthread.h>
#include <unistd.h>
#include <stdio.h>

typedef struct _comm_job
{
    struct _comm_job * prev;
    struct _comm_job * next;

    thread_pool * pool;
    thread_pool_work_func func;
    void * request;
} _comm_job;

typedef struct _comm_thread {
    // link to global list
    struct _comm_thread * prev;
    struct _comm_thread * next;

    // link to pool list
    struct _comm_thread * w_prev;
    struct _comm_thread * w_next;

    pthread_t id;

    _comm_job * job;
} _comm_thread;

struct thread_pool
{
    struct thread_pool * prev;
    struct thread_pool * next;

    void * ctx;
    int max;
    int cur;

    _comm_thread * work_thread;
};

static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t  cond  = PTHREAD_COND_INITIALIZER;

static void lock()
{
    pthread_mutex_lock(&mutex);
}

static void unlock()
{
    pthread_mutex_unlock(&mutex);
}

static void wait_()
{
    pthread_cond_wait(&cond, &mutex);
}

static void wakeup()
{
    pthread_cond_signal(&cond);
}

static _comm_thread * thread_list = 0;
static thread_pool * pool_list = 0;
static int max_thread = 0;
static int cur_thread = 0;
static int idle_thread = 0;
static _comm_job * job_list = 0;

thread_pool * thread_pool_create(int max, void * ctx)
{
    thread_pool * pool = GET_OBJECT(thread_pool);
    if (pool) {
        pool->max = max;
        pool->ctx = ctx;
        pool->cur = 0;
        pool->work_thread = 0;

        lock();
        max_thread += max;
        dlist_insert_tail(pool_list, pool);
        unlock();
    }
    return pool;
}

#define UNLOCK_DO(work) \
    do { unlock(); work; lock(); } while(0)

void thread_pool_destory(thread_pool * pool)
{
    lock();
    if (pool) {
        max_thread -= pool->max;

        //wait fo exit
        while (pool->cur) {
            UNLOCK_DO(usleep(10 * 1000));
        }
    }

    if (cur_thread > max_thread) {
        pthread_cond_broadcast(&cond);
    }

    dlist_remove(pool_list, pool);
    unlock();

    WRITE_DEBUG_LOG("thread_pool_destory %d/%d", cur_thread, max_thread);
    if (pool) {
        RELEASE_OBJECT(thread_pool, pool);
    }
}
void thread_pool_safe_exit()
{
    lock();
    while (pool_list) {
        UNLOCK_DO(thread_pool_destory(pool_list));
    }
    
    while(cur_thread > 0) {
        UNLOCK_DO(usleep(10 * 1000));
    }
    unlock();
}

static void * work_proc(void * p)
{
    pthread_detach(pthread_self());

    _comm_thread * t = (_comm_thread*)p;
    while (1) {
        // wait
        lock();
        while(job_list == 0) {
            // exit
            if (cur_thread > max_thread) {
                dlist_remove(thread_list, t);
                cur_thread--;
                assert(cur_thread >= 0);
                idle_thread--;
                assert(idle_thread >= 0);
                unlock();
                goto exit;
            }

            WRITE_DEBUG_LOG("-- 0x%016lx %p wait",
                    (unsigned long)pthread_self(), t);
            wait_();
            WRITE_DEBUG_LOG("-- 0x%016lx %p wakeup",
                    (unsigned long)pthread_self(), t);
        }
        _comm_job * j = job_list;
        dlist_remove(job_list, j);
        dlist_insert_tail_x(j->pool->work_thread, t, w_prev, w_next);
        unlock();

        assert(j->pool);
        assert(j->func);

        // do job
        t->job = j;
        if (j->pool && j->func) {
            j->func(j->request, j->pool->ctx);
        }
        t->job = 0;

        // release
        lock();
        if (j->pool) {
            dlist_remove_x(j->pool->work_thread, t, w_prev, w_next);
            j->pool->cur--;
            assert(j->pool->cur >= 0);
        }
        idle_thread++;
        unlock();

        RELEASE_OBJECT(_comm_job, j);

        WRITE_DEBUG_LOG("-- 0x%016lx %p release %d/%d",
                (unsigned long)pthread_self(), t,
                cur_thread, max_thread);
    }

exit:
    // free and exit
    RELEASE_OBJECT(_comm_thread, t);

    WRITE_DEBUG_LOG("-- 0x%016lx %p exit %d/%d",
            (unsigned long)pthread_self(), t,
            cur_thread, max_thread);
    return 0;
}

static _comm_thread * create_thread()
{
    _comm_thread * t = GET_OBJECT(_comm_thread);
    t->prev = t->next = 0;
    t->w_prev = t->w_next = 0;
    t->id = 0;
    t->job = 0;

    if (pthread_create(&t->id, 0, work_proc, t) != 0) {
        goto failed;
    }

    WRITE_DEBUG_LOG("-- 0x%016lx %p created", (unsigned long)t->id, t);
    return t;

failed:
    if (t) {
        if (t->id) {
            pthread_cancel(t->id);
        }
        RELEASE_OBJECT(_comm_thread, t);
    }
    return 0;
}

static int push_job(_comm_job * job)
{
    assert( job && job->pool);

    thread_pool * pool = job->pool;
    lock();
    if (pool->cur >= pool->max) {
        unlock();
        return -1;
    }

    if (idle_thread > 0) {
        idle_thread--;
        assert(idle_thread >= 0);
        pool->cur ++;
        assert(pool->cur <= pool->max);
        dlist_insert_tail(job_list, job);
        unlock();
        return 0;
    }

    if (cur_thread >= max_thread) {
        unlock();
        return -1;
    }
    cur_thread++;
    unlock();

    _comm_thread * t = create_thread();

    if (!t) {
        lock();
        cur_thread--;
        unlock();
        return -1;
    }

    lock();
    dlist_insert_tail(thread_list, t);
    pool->cur ++;
    assert(pool->cur <= pool->max);
    dlist_insert_tail(job_list, job);
    unlock();
    return 0;
}

int thread_pool_do(thread_pool * pool,
                   thread_pool_work_func func,
                   void * request)
{
    if (pool == 0 || func == 0) {
        return -1;
    }

    _comm_job * job = GET_OBJECT(_comm_job);
    if (job == 0) {
        return -1;
    }

    job->pool = pool;
    job->func = func;
    job->request = request;
    if (push_job(job) != 0) {
        RELEASE_OBJECT(_comm_job, job);
        return -1;
    }

    WRITE_DEBUG_LOG("-- signaled");
    wakeup();
    return 0;
}
